import org.pentaho.di.trans.steps.userdefinedjavaclass.TransformClassBase;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.core.exception.KettleException;
import org.apache.commons.net.ftp.*;
import java.io.*;

public class FTPRecursiveDownloader extends TransformClassBase {

    // 必须实现的processRow方法
    @Override
    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        // 首先调用父类方法处理行
        Object[] r = getRow();
        if (r == null) {
            setOutputDone();
            return false;
        }

        if (first) {
            first = false;
            // 初始化操作可以放在这里
        }

        // 获取参数（从字段或变量获取）
        String server = getVariable("FTP_SERVER", "192.168.1.4");
        int port = Integer.parseInt(getVariable("FTP_PORT", "21"));
        String user = getVariable("FTP_USER", "alpineftp_user");
        String pass = getVariable("FTP_PASSWORD", "alpineftp_password");
        String remotePath = getVariable("REMOTE_PATH", "/ftp/alpineftp_user/2025-01");
        String localPath = getVariable("LOCAL_PATH", "/home/tomcat/.kettle/data/test");

        // 创建FTP客户端
        FTPClient ftp = new FTPClient();

        try {
            // 连接FTP服务器
            logBasic("正在连接FTP服务器: " + server + ":" + port);
            ftp.connect(server, port);
            if (!ftp.login(user, pass)) {
                throw new KettleException("FTP登录失败");
            }

            // 设置传输模式
            ftp.enterLocalPassiveMode();
            ftp.setFileType(FTP.BINARY_FILE_TYPE);

            // 开始递归下载
            logBasic("开始下载目录: " + remotePath);
            downloadDirectory(ftp, remotePath, localPath);

            // 登出并断开连接
            ftp.logout();
            logBasic("FTP操作完成");

        } catch (Exception e) {
            logError("FTP操作失败", e);
            throw new KettleException("FTP操作失败", e);
        } finally {
            try {
                if (ftp.isConnected()) {
                    ftp.disconnect();
                }
            } catch (IOException e) {
                logError("断开FTP连接时出错", e);
            }
        }

        // 将输入行原样传递到输出（如需修改可在此处理）
        putRow(getInputRowMeta(), r);
        return true;
    }

    // 递归下载目录方法
    private void downloadDirectory(FTPClient ftp, String remoteDir, String localDir) throws IOException {
        // 创建本地目录
        File localDirFile = new File(localDir);
        if (!localDirFile.exists() && !localDirFile.mkdirs()) {
            throw new IOException("无法创建本地目录: " + localDir);
        }

        // 获取远程文件列表
        FTPFile[] files = ftp.listFiles(remoteDir);
        if (files == null) {
            logError("无法获取目录列表: " + remoteDir);
            return;
        }

        for (FTPFile file : files) {
            if (file == null) continue;

            String remoteFilePath = remoteDir + "/" + file.getName();
            String localFilePath = localDir + File.separator + file.getName();

            if (file.isDirectory()) {
                // 递归处理子目录
                downloadDirectory(ftp, remoteFilePath, localFilePath);
            } else {
                // 下载文件
                logBasic("正在下载: " + remoteFilePath);

                File localFile = new File(localFilePath);
                if (localFile.exists()) {
                    logBasic("文件已存在，跳过: " + localFilePath);
                    continue;
                }

                OutputStream output = null;
                try {
                    output = new BufferedOutputStream(new FileOutputStream(localFile));
                    if (!ftp.retrieveFile(remoteFilePath, output)) {
                        throw new IOException("FTP服务器返回失败状态");
                    }
                    logBasic("成功下载: " + localFilePath);
                } catch (IOException e) {
                    logError("下载文件失败: " + remoteFilePath, e);
                    if (localFile.exists()) {
                        localFile.delete();
                    }
                } finally {
                    if (output != null) {
                        try {
                            output.close();
                        } catch (IOException e) {
                            logError("关闭文件流时出错", e);
                        }
                    }
                }
            }
        }
    }
}